#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "mq_consumer.hpp"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <condition_variable>

namespace mymq
{
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr; // connection里面会用到
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
        : _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec)
        {}
        ~Channel()
        {
            basicCancel();
        }

        std::string getCid()
        {
            return _cid;
        }

        // 服务端也要有与客户端对应的信道
        bool openChannel()
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid();
            openChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            // 发送
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr res = waitResponse(rid);
            // 返回
            return res->ok();
        }
        void closeChannel()
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid();
            closeChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            // 发送
            _codec->send(_conn, req);
            // 等待响应
            waitResponse(rid);
        }

        bool declareExchange(const std::string& name, ExchangeType type, bool durable, 
            bool auto_delete, google::protobuf::Map<std::string, std::string>& args)
        {
            // 1.构建请求对象
            std::string rid = UUIDHelper::uuid();
            declareExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            req.set_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 2.发送请求对象
            _codec->send(_conn, req);
            // 3.等待响应
            basicCommonResponsePtr res = waitResponse(rid);
            // 4.返回
            return res->ok();
        }
        void deleteExchange(const std::string& name)
        {
            // 1.构建请求对象
            std::string rid = UUIDHelper::uuid();
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            // 2.发送请求对象
            _codec->send(_conn, req);
            // 3.等待响应
            waitResponse(rid);
        }

        bool declareQueue(const std::string& qname, bool qdurable, bool qexclusive, bool qauto_delete,
            google::protobuf::Map<std::string, std::string>& qargs)
        {
            // 1.构建请求对象
            std::string rid = UUIDHelper::uuid();
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_durable(qdurable);
            req.set_exclusive(qexclusive);
            req.set_auto_delete(qauto_delete);
            req.mutable_args()->swap(qargs);
            // 2.发送请求对象
            _codec->send(_conn, req);
            // 3.等待响应
            basicCommonResponsePtr res = waitResponse(rid);
            // 4.返回
            return res->ok();
        }
        void deleteQueue(const std::string& qname)
        {
            // 1.构建请求对象
            std::string rid = UUIDHelper::uuid();
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            // 2.发送请求对象
            _codec->send(_conn, req);
            // 3.等待响应
            waitResponse(rid);
        }

        bool queueBind(const std::string& ename, const std::string& qname, const std::string& key)
        {
            // 构造请求
            std::string rid = UUIDHelper::uuid();
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(key);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr res = waitResponse(rid);
            // 返回
            return res->ok();
        }
        void queueUnBind(const std::string& ename, const std::string& qname)
        {
            // 1.构建请求对象
            std::string rid = UUIDHelper::uuid();
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            // 2.发送请求对象
            _codec->send(_conn, req);
            // 3.等待响应
            waitResponse(rid);
        }

        bool basicPublish(const std::string& ename, const BasicProperties* bp, const std::string& body)
        {
            // 构造请求
            std::string rid = UUIDHelper::uuid();
            basicPublishRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_body(body);
            // 属性不一定存在
            if (bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }

            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr res = waitResponse(rid);
            // 返回
            return res->ok();
        }
        // 确认某队列的某消息，可以通过消费者找到订阅的队列
        void basicAck(const std::string& msg_id)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("消息确认时，找不到消费者信息！");
                return;
            }

            // 构建请求
            std::string rid = UUIDHelper::uuid();
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_msg_id(msg_id);
            // 发送
            _codec->send(_conn, req);
            // 等待
            waitResponse(rid);
        }

        bool basicConsume(const std::string& ctag, const std::string& qname, bool auto_ack,
            const ConsumerCallback& cb)
        {
            // 检查消费者是否已经存在
            if (_consumer.get() != nullptr)
            {
                DLOG("当前信道已经订阅其他队列消息！");
                return false;
            }

            // 构建请求
            std::string rid = UUIDHelper::uuid();
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(ctag);
            req.set_queue_name(qname);
            req.set_auto_ack(auto_ack);
            // 发送
            _codec->send(_conn, req);
            // 等待
            basicCommonResponsePtr res = waitResponse(rid);
            if (res->ok() == false)
            {
                DLOG("添加订阅失败！");
                return false;
            }
            // 订阅成功，信道的角色变为消费者。需要构建一个消费者并关联
            _consumer = std::make_shared<Consumer>(ctag, qname, auto_ack, cb);
            return true;
        }
        // 取消订阅需要指定 队列和消费者标识，这两个信息可以从消费者中得到
        void basicCancel()
        {
            if (_consumer.get() == nullptr)
            {
                return;
            }

            // 构建请求
            std::string rid = UUIDHelper::uuid();
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_consumer_tag(_consumer->tag);
            // 发送
            _codec->send(_conn, req);
            // 等待
            waitResponse(rid);
            _consumer.reset();
        }
    public:
        // 当连接收到基础响应时，将响应添加到信道的 hash_map
        void putBasicResponse(const basicCommonResponsePtr& resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(), resp));
            _cv.notify_all();
        }
        // 当连接收到消息推动后，到信道中寻找消费者，调用消费者的回调函数进行消息处理
        void consume(const basicConsumeResponsePtr& resp)
        {
            // 检查是否拥有消费者
            if (_consumer.get() == nullptr)
            {
                DLOG("消息处理时，未找到订阅者信息！");
                return;
            }

            // 检查推送消息的消费者标识和当前消费者的标识是否一致
            if (_consumer->tag != resp->consumer_tag())
            {
                DLOG("服务器推送的消息中的消费者，与当前信道的消费者不一致！");
                return;
            }

            // 消费者没问题，调用消费者的回调函数对消息进行处理
            _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }
    private:
        // 等待响应
        basicCommonResponsePtr waitResponse(const std::string& rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [&rid, this](){
                // 唤醒条件：请求被服务端处理，客户端收到响应
                return _basic_resp.find(rid) != _basic_resp.end();
            });
            basicCommonResponsePtr resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return resp;
        }
    private:
        std::string _cid;
        muduo::net::TcpConnectionPtr _conn; // 信道关联的网络通信连接
        ProtobufCodecPtr _codec; // protobuf 协议处理句柄
        Consumer::ptr _consumer; // 关联的消费者
        // 请求id-响应 hash表，用来快速查询某个请求是否已经完成并接收到响应
        std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;
        // 因为 muduo 的操作是异步的，所以当信道发送请求（例如声明交换机），不一定能立刻完成。所以需要条件变量进行同步操作
        // 只有当请求被完成，收到响应后才进行其他操作：例如发送消息等
        std::mutex _mutex;
        std::condition_variable _cv;
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        Channel::ptr create(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 创建新的channel，插入 信道id-信道 映射关系
            Channel::ptr newchannel = std::make_shared<Channel>(conn, codec);
            _channels.insert(std::make_pair(newchannel->getCid(), newchannel));
            
            return newchannel;
        }

        void remove(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }

        Channel::ptr get(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
                return Channel::ptr();
            return it->second;
        }
    private:
        std::mutex _mutex;
        // 信道id-信道
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}

#endif